package com.wanxing.sellerapp.emqx

import android.content.Context
import android.util.Log
import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.JSONObject
import com.blankj.utilcode.util.SPUtils
import com.blankj.utilcode.util.ThreadUtils
import com.wanxing.sellerapp.App
import com.wanxing.sellerapp.config.AppConfig
import com.wanxing.sellerapp.config.EmqxConfig
import com.wanxing.sellerapp.emqx.entity.EmqMessage
import com.wanxing.sellerapp.emqx.entity.EmqPayload
import com.wanxing.sellerapp.emqx.entity.MustArriveMsg
import com.wanxing.sellerapp.emqx.enums.ActionEnum
import com.wanxing.sellerapp.emqx.interfaces.IMessenger
import com.wanxing.sellerapp.platform.Api
import com.wanxing.sellerapp.platform.entity.Seller
import com.wanxing.sellerapp.seller.*
import com.wanxing.sellerapp.seller.entity.ScaleWeight
import com.wanxing.sellerapp.seller.events.AfterRegisterEvent
import com.wanxing.sellerapp.seller.events.ConnectChangeEvent
import com.wanxing.sellerapp.seller.sdk.video.VideoSellerCallback
import com.wanxing.sellerapp.util.RandomUtil
import org.eclipse.paho.android.service.MqttAndroidClient
import org.eclipse.paho.client.mqttv3.*
import org.greenrobot.eventbus.EventBus
import java.lang.Exception
import java.util.concurrent.LinkedBlockingQueue

/**
 * mid: 机器号
 */
class MyMqttClient(private val context: Context,val mSellerManager:SellerManager ):IMessenger {
    

    /**
     * 存储还未到发送成功的必达消息。
     */
    private val mustArriveMsg = LinkedBlockingQueue<MustArriveMsg>(256)

    private val actionDelegator = ActionDelegator(this, mSellerManager)

    private var currentMsgId = ""

    private val reSendThread = Thread({
        while (true){
            try {
                if (mqttClient.isConnected){
                    val message = mustArriveMsg.take()
                    publish(message.topic, message.data, true, message.qos, message.isRetain)
                    Thread.sleep(300)
                } else{
                    Thread.sleep(1000)
                }

            } catch (e: InterruptedException){
                break
            }
        }
    },"reSendThread")

    private lateinit var mqttClient: MqttAndroidClient


    fun init() {
       ThreadUtils.executeByIo(object :ThreadUtils.Task<Seller>(){
           override fun doInBackground(): Seller {
               var seller: Seller?
               do {
                   seller = Api.register()
               } while (seller == null)
               return seller
           }
           override fun onSuccess(result: Seller) {
               EventBus.getDefault().post(AfterRegisterEvent(result))
               App.mid = result.mid
               SPUtils.getInstance().put("device-mid", result.mid)
               connectEmq()
               initSeller()
           }
           override fun onCancel() {}
           override fun onFail(t: Throwable?) {}
       })

    }

    private fun initSeller(){
        reSendThread.start()
        mSellerManager.open(context, SellerEventCallback(this, mSellerManager),
            object :VideoSellerCallback{
            override fun onRepeatUnlock(lockNo: Int) {}
            override fun onOpen(lockNo: Int, type: Int) {
                val msg = EmqMessage()
                msg.mid = App.mid
                msg.messageId = currentMsgId
                msg.action = ActionEnum.OPEN_DOOR
                msg.doorNo = lockNo
                val json = JSONObject()
                json["type"] = type
                msg.data= json
                sendMsgToService(JSON.toJSONString(msg), true,2)
            }

            override fun onClose(lockNo: Int, type: Int) {
                val msg = EmqMessage()
                msg.mid = App.mid
                msg.messageId = currentMsgId
                msg.action = ActionEnum.CLOSE_DOOR
                msg.doorNo = lockNo
                val json = JSONObject()
                json["type"] = type
                msg.data= json
                sendMsgToService(JSON.toJSONString(msg), true,2)
            }
        })
    }

    private fun connectEmq() {
        val clientId = "${App.mid}-${RandomUtil.randomString(2)}"
        mqttClient = MqttAndroidClient(context, AppConfig.brokerUrl, clientId)
        mqttClient.registerResources(context)
        mqttClient.setCallback(object : MqttCallbackExtended {
            override fun messageArrived(topic: String?, message: MqttMessage?) {
                message?.let {
                    val msg = it.toString()
                    Log.i(AppConfig.LOGTAG, "receive message = $msg")
                    try {
                        val payload = JSON.parseObject(msg, EmqPayload::class.java)
                        currentMsgId = payload.messageId
                        actionDelegator.delegate(payload)
                    } catch (e: Exception) {
                        Log.e(AppConfig.LOGTAG, "消息格式错误")
                    }
                }
            }

            override fun connectionLost(cause: Throwable?) {
                Log.i(AppConfig.LOGTAG, "Connection lost, cause: ${cause.toString()}")
                recordDisconnect()
                // 丢失连接保存到propertis
            }


            override fun deliveryComplete(token: IMqttDeliveryToken) {
//                Log.i(AppConfig.LOGTAG, "DeliveryComplete ${token.message}")
            }

            override fun connectComplete(reconnect: Boolean, serverURI: String?) {
                recordConnect()
                subscribe(topic = "seller/${App.mid}/api", qos = 2)
            }
        })
        doConnect()
    }

    private fun recordDisconnect(){
        EventBus.getDefault().post(ConnectChangeEvent(false))
        SPUtils.getInstance().put("mq_conn", false)
    }

    private fun recordConnect(){
        EventBus.getDefault().post(ConnectChangeEvent(true))
        SPUtils.getInstance().put("mq_conn", true)
    }

    private fun doConnect(){
        val options = MqttConnectOptions()
        options.isAutomaticReconnect = true
        options.isCleanSession = true
        options.userName = EmqxConfig.connUsername
        options.password = AppConfig.connPassword.toCharArray()
        options.connectionTimeout = 8
        try {
            mqttClient.connect(options, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    Log.i(AppConfig.LOGTAG, "连接 broker 成功")
                    subscribe(topic = "seller/${App.mid}/api", qos = 2)
                    recordConnect()
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.e(AppConfig.LOGTAG, "连接broker失败, errorMsg = ${exception?.message}")

                    exception?.message?.let {
                        if (it == "已连接客户机"){ return@onFailure }
                        else{
                            reConnect()
                            recordDisconnect()
                        }
                    }
                }
            })
        } catch (e: MqttException) {
            Log.e(AppConfig.LOGTAG, "连接broker异常, errorMsg = ${e.message}")
            reConnect()
        }
    }

    private fun reConnect() {
        mqttClient.unregisterResources()
        connectEmq()
    }

    private fun sendCurrentDoorWeight(scaleWeightList: List<ScaleWeight>, doorNo: Int){
        val data = JSONObject()
        data["type"] = "DOOR_CURRENT_WEIGHTS"
        data["data"] = mapOf("doorNo" to doorNo, "weights" to scaleWeightList)
        publish("seller/${App.mid}/push", data.toJSONString())
    }



    private fun subscribe(topic: String, qos: Int = 1) {
        try {
            mqttClient.subscribe(topic, qos, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    //TODO 记录订阅成功日志
                    recordConnect()
                    Log.i(AppConfig.LOGTAG, "订阅主题： $topic 成功")
                }
                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.e(AppConfig.LOGTAG, "订阅主题： $topic 失败 errMsg = ${exception?.message}")
                    //TODO 记录订阅失败日志
                }
            })
        } catch (e: MqttException) {
            Log.e(AppConfig.LOGTAG, "订阅主题： $topic 异常 errMsg = ${e.message}")
            //TODO 记录订阅失败日志
        }
    }

    fun unsubscribe(topic: String) {
        mqttClient.unsubscribe(topic, null, object : IMqttActionListener {
            override fun onSuccess(asyncActionToken: IMqttToken?) {
                Log.i(AppConfig.LOGTAG, "Unsubscribed to $topic")
            }
            override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                Log.e(AppConfig.LOGTAG, "Failed to unsubscribe $topic")
            }
        })
    }

    fun publish(topic: String, msg: String, mustArrive: Boolean = false,  qos: Int = 1, retained: Boolean = false, ) {
        try {
            val message = MqttMessage()
            message.payload = msg.toByteArray()
            message.qos = qos
            message.isRetained = retained
            if (mustArrive && !mqttClient.isConnected) {
                mustArriveMsg.offer(MustArriveMsg(msg, topic, qos, retained))
                return
            }
            mqttClient.publish(topic, message, null, object : IMqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    Log.i(AppConfig.LOGTAG, "$msg published to $topic")
                }
                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.i(AppConfig.LOGTAG, "Failed to publish $msg to $topic")
                    if (mustArrive){
                        mustArriveMsg.offer(MustArriveMsg(msg, topic, qos, retained))
                    }
                }
            })
        } catch (e: MqttException) {
            Log.e(AppConfig.LOGTAG, "Publish Error", e)
            if (mustArrive){
                mustArriveMsg.offer(MustArriveMsg(msg, topic, qos, retained))
            }
        }
    }

    fun release() {
        mSellerManager.close()
        reSendThread.interrupt()
        mqttClient.disconnect(null, object : IMqttActionListener {
            override fun onSuccess(asyncActionToken: IMqttToken?) {
                Log.d(AppConfig.LOGTAG, "Disconnected")
            }
            override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                Log.d(AppConfig.LOGTAG, "Failed to disconnect")
            }
        })
        mqttClient.unregisterResources()
    }

    override fun sendMsgToService(json: String,mustArrive: Boolean,  qos: Int,) {
        publish(EmqxConfig.pushServiceTopic, json, mustArrive, qos)
    }

    override fun sendMsgToWeb(topic: String, json: String) {
        publish(topic, json)
    }

    override fun clearRetainMessage() {
        publish("seller/${App.mid}/api", "", false, 0, true)
    }
}